package com.itzx.marketanalysis

import java.sql.Timestamp
import java.util.UUID
import java.util.concurrent.TimeUnit

import org.apache.flink.api.common.functions.AggregateFunction
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.util.Random

/**
 *
 *
 * author: yyeleven
 * create: 2020/3/23 21:43
 */
object AppMarketing {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    env.setParallelism(1)

    val dataStream = env.addSource(new SimulatedEventSource_2())
      .assignAscendingTimestamps(_.timestamp)
      .filter(_.behavior != "UNINSTALL")
      .map(data => {
        ("dummyKey", 1L)
      }).keyBy(_._1)
      .timeWindow(Time.hours(1), Time.seconds(10))
      .aggregate(new CountAgg(), new MarketingCountTotal())

    dataStream.print()

    env.execute("AppMarketing")
  }
}

class SimulatedEventSource_2 extends RichSourceFunction[MarketingUserBehavior] {

  // 定义是否运行的标志位
  var running = true
  // 定义用户行为的集合
  val behaviorTypes: Seq[String] = Seq("CLICK", "DOWNLOAD", "INSTALL", "UNINSTALL")
  // 定义渠道的集合
  val channelSets: Seq[String] = Seq("wechat", "weibo", "appstore", "huaweistore")
  // 定义一个随机数发生器
  val random: Random = new Random()

  override def run(ctx: SourceFunction.SourceContext[MarketingUserBehavior]): Unit = {

    // 定义一个生成数据的上限
    val maxElements: Long = Long.MaxValue
    var count = 0L

    // 随机生成所有数据
    while(running && count < maxElements) {
      val id: String = UUID.randomUUID().toString
      val behavior: String = behaviorTypes(random.nextInt(behaviorTypes.size))
      val channel: String = channelSets(random.nextInt(channelSets.size))
      val timestamp: Long = System.currentTimeMillis()

      ctx.collect(MarketingUserBehavior(id, behavior, channel, timestamp))
      count += 1L
      TimeUnit.MICROSECONDS.sleep(10L)
    }
  }

  override def cancel(): Unit = {
    running = false
  }
}

class CountAgg() extends AggregateFunction[(String, Long), Long, Long] {
  override def createAccumulator(): Long = 0L

  override def add(value: (String, Long), accumulator: Long): Long = accumulator + 1L

  override def getResult(accumulator: Long): Long = accumulator

  override def merge(a: Long, b: Long): Long = a + b
}

class MarketingCountTotal() extends WindowFunction[Long, MarketingViewCount, String, TimeWindow] {
  override def apply(key: String, window: TimeWindow, input: Iterable[Long], out: Collector[MarketingViewCount]): Unit = {

    val windowStart: String = new Timestamp(window.getStart).toString
    val windowEnd: String = new Timestamp(window.getEnd).toString
    val count: Long = input.iterator.next()
    out.collect(MarketingViewCount(windowStart, windowEnd, "appMarketing", "total", count))
  }
}


